package spark.session;

import org.apache.spark.util.AccumulatorV2;

import java.util.HashMap;

public class SessionAggrStatAccumulator extends AccumulatorV2<String,HashMap<String,Integer>>  {

    private HashMap<String,Integer> aggrMap = new HashMap<>();
    @Override
    //当accumulateV2中存在数据不存在这种问题，是否结束程序
    public boolean isZero() {
        return aggrMap.isEmpty();
    }

    @Override
    public AccumulatorV2<String, HashMap<String, Integer>> copy() {
        SessionAggrStatAccumulator newMyAccu = new SessionAggrStatAccumulator();
        newMyAccu.aggrMap = this.aggrMap;
        return newMyAccu;
    }

    @Override
    public void reset() {
        aggrMap.clear();

    }

    @Override
    public void add(String v) {
        aggrMap.put(v,aggrMap.getOrDefault(v, 0)+1);
    }

    @Override
    public void merge(AccumulatorV2<String, HashMap<String, Integer>> other) {
        if(other instanceof SessionAggrStatAccumulator){
            other.value().forEach((k,v)->{
                this.aggrMap.put(k, other.value().get(k)+this.aggrMap.getOrDefault(k,0));
            }) ;
        }
    }

    @Override
    public HashMap<String, Integer> value() {
        return this.aggrMap;
    }
}

